package com.launch.timer;



import javax.security.auth.callback.LanguageCallback;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;

/**
 * @author ennian
 *
 * 时间轮操作实例
 */
public class LaunchWheelTimer implements Timer{


    public static final int STATE_INIT = 0 ; //时间论开始
    public static final int STATE_STARTED = 1 ;  //时间论初始化
    public static final int STATE_SHUTDOWN = 2 ;  //时间论关闭

    public volatile int state = STATE_INIT ;

    //原子引用 修改对象属性
    AtomicIntegerFieldUpdater stateUpdater = AtomicIntegerFieldUpdater.newUpdater(LaunchWheelTimer.class,"state") ;

    private final CountDownLatch startTimeInitialized = new CountDownLatch(1) ;

    private volatile long startTime;

    // 时间轮
    private final LaunchWheelBucket[] wheel ;

    private final int mask;

    private final long tickDuration ;

    private final Worker worker = new Worker();

    private final Thread workerThread;

    // 保证单例
    private static class LaunchWheelTimerHandler {
        private static LaunchWheelTimer instance = new LaunchWheelTimer() ;
    }

    // 懒汉式 单例模式 
    public static LaunchWheelTimer getInstance(){
        return LaunchWheelTimerHandler.instance;
    }

    //取消任务队列
    public final BlockingDeque<LaunchWheelTimeout> cancelTimeOutQueue = new LinkedBlockingDeque<>()  ;

    public final BlockingDeque<LaunchWheelTimeout> timeouts = new LinkedBlockingDeque<>()  ;


    private LaunchWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit, int ticksPerWheel) {
        if (threadFactory == null) {
            throw new IllegalArgumentException("launch exception ======>threadFactory null is wrong ");
        }
        if (unit == null) {
            throw new IllegalArgumentException("launch exception =======>unit null is wrong ");
        }
        if (tickDuration <= 0) {
            throw new IllegalArgumentException("launch exception ====> tickDuration must be greater than 0: " + tickDuration);
        }
        if (ticksPerWheel <= 0) {
            throw new IllegalArgumentException("launch exception =====> ticksPerWheel must be greater than 0: " + ticksPerWheel);
        }

        // 初始化 时间论 ，以 2 的 n次幂 扩容
        wheel = createWheel(ticksPerWheel);
        // 用于存放桶
        mask = wheel.length - 1;

        // 转换成纳秒 .
        this.tickDuration = unit.toNanos(tickDuration);

        // 是否溢出判断 .
        if (this.tickDuration >= Long.MAX_VALUE / wheel.length) {
            throw new IllegalArgumentException(String.format("tickDuration: %d (expected: 0 < tickDuration in nanos < %d", tickDuration, Long.MAX_VALUE / wheel.length));
        }
        workerThread = threadFactory.newThread(worker);
    }

    //默认 512个桶
    private LaunchWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit) {
        this(threadFactory, tickDuration, unit, 512);
    }
    // 512个桶 100毫秒转动一次
    private LaunchWheelTimer(ThreadFactory threadFactory) {
        this(threadFactory, 100, TimeUnit.MILLISECONDS);
    }

    private LaunchWheelTimer(long tickDuration, TimeUnit unit, int ticksPerWheel) {
        this(Executors.defaultThreadFactory(), tickDuration, unit, ticksPerWheel);
    }

    private LaunchWheelTimer(long tickDuration, TimeUnit unit) {
        this(Executors.defaultThreadFactory(), tickDuration, unit);
    }
    //使用默认线程池
    private LaunchWheelTimer() {
        this(Executors.defaultThreadFactory());
    }
    /**
     *  创建时间论
     * @param ticksPerWheel 一圈有几个bucket
     * @return
     */
    private static LaunchWheelBucket[] createWheel(int ticksPerWheel){
        if (ticksPerWheel<= 0 )
        throw new IllegalArgumentException("ticksPerWheel must be greater than 0: " + ticksPerWheel) ;
        if (ticksPerWheel > 1073741824) {
            throw new IllegalArgumentException(
                    "ticksPerWheel may not be greater than 2^30: " + ticksPerWheel);
        }
        // 初始化轮子数 ,二次幂扩容 2
        //4
        //8
        //16
        //32
        //64
        int initPerWheel = 1;
        while (initPerWheel < ticksPerWheel) {
            initPerWheel <<= 1;
        }
        //时间论数组
        LaunchWheelBucket[] wheel = new LaunchWheelBucket[initPerWheel] ;
        for (int i = 0 ; i<wheel.length ; i++)
            wheel[i] = new LaunchWheelBucket() ;
        return wheel ;
    }

    /**
     * 时间论启动方法
     */
    public void start(){
        final int i = stateUpdater.get(this);
        // 如果当前时间论是初始化状态
        switch (i) {
            case STATE_INIT : {
                // 更新初始化状态 dcl
                if (i == STATE_INIT && stateUpdater.compareAndSet(this, STATE_INIT, STATE_STARTED)) {
                    workerThread.start();
                    break;
                }
            } case STATE_STARTED:
                break;
            case STATE_SHUTDOWN: {
                throw new IllegalStateException("launch exception ====>can not start , because it has been shutdown");
            } default: {
                throw new IllegalStateException("launch exception ====> wheel unknow state ");
            }
        }
        // 等待直到开始时间被初始化
        while (startTime == 0) {
            try {
                startTimeInitialized.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }


    public final class Worker implements Runnable{

        private final Set<TimeOut> unProcessedTimeOuts = new HashSet<>() ;

        private long tick;

        @Override
        public void run() {
            //开始时间
            startTime = System.nanoTime();
            // 0 作为没有被初始化时的标志 ， 初始化开始时间
            if (startTime == 0) startTime = 1 ;
            // 通知其它线程初始化完成
            startTimeInitialized.countDown();
            do {
                final long deadline = waitForNewTick();
                if (deadline>0){
                    // 获取时间论桶索引
                    int index = (int) (tick & mask) ;
                    runTheCancelTask();
                    final LaunchWheelBucket bucket = wheel[index];
                    processTimeOutToBucket();
                    bucket.expireTimeouts(deadline);
                    tick++ ;
                }
            }while (stateUpdater.get(LaunchWheelTimer.this) == STATE_STARTED) ;

            for (LaunchWheelBucket bucket: wheel) {
                bucket.clearTimeOuts(unProcessedTimeOuts);
            }

            for (;;) {
                LaunchWheelTimeout timeout = timeouts.poll();
                if (timeout == null) {
                    break;
                }
                if (!timeout.isCancelled()) {
                    unProcessedTimeOuts.add(timeout);
                }
            }
            runTheCancelTask();
        }

        private void processTimeOutToBucket() {
            for (int i = 0; i < 100000; i++) {
                LaunchWheelTimeout timeout = timeouts.poll();
                if (timeout == null) {
                    //所有都处理完成
                    break;
                }
                //如果已经取消状态直接退出
                if (timeout.state == LaunchWheelTimeout.STATE_CANCELED) {
                    continue;
                }
                int finalIndex = getFinalIndex(timeout);
                LaunchWheelBucket bucket = wheel[finalIndex];
                bucket.addTimeOut(timeout);
            }
        }

        private int getFinalIndex(LaunchWheelTimeout timeout) {
            //获取 bucket索引下标
            long calculated = timeout.deadline / tickDuration;
            timeout.remainingRounds = (calculated - tick) / wheel.length;
            final long ticks = Math.max(calculated, tick);
            int stopIndex = (int) (ticks & mask);
            return stopIndex;
        }

        private void runTheCancelTask(){
            for (;;){
                final LaunchWheelTimeout poll = cancelTimeOutQueue.poll();
                if (poll == null){
                    break;
                }
                poll.remove();
            }
        }

        /**
         * 计算目标开始时间和当前tick数量，然后等到目标时间到达
         * @return
         */
        private long waitForNewTick(){
            long deadLine = tickDuration * (tick + 1);

            for (;;){
                long currentTime = System.nanoTime() - startTime;
                long sleepTimeMs = (deadLine - currentTime + 999999) / 1000000;
                if (sleepTimeMs<=0){
                    if (currentTime == Long.MIN_VALUE) return -Long.MIN_VALUE ;
                    else return currentTime ;
                }
            }
        }
        public Set<TimeOut> unprocessedTimeouts() {
            return Collections.unmodifiableSet(unProcessedTimeOuts);
        }
    }

    @Override
    public TimeOut newTimeOut(TimerTask timerTask, long delay, TimeUnit timeUnit) {
        if (timerTask == null) {
            throw new NullPointerException("timerTask");
        }
        if (timeUnit == null) {
            throw new NullPointerException("timeUnit");
        }
        start();
        long deadline = System.nanoTime() + timeUnit.toNanos(delay) - startTime;
        LaunchWheelTimeout timeout = new LaunchWheelTimeout(this, timerTask, deadline);
        timeouts.add(timeout);
        return timeout;
    }





    @Override
    public Set<TimeOut> stop() {
        if (Thread.currentThread() == workerThread) {
            throw new IllegalStateException(LaunchWheelTimeout.class.getSimpleName() + ".stop() cannot be called from " + TimerTask.class.getSimpleName());
        }

        if (!stateUpdater.compareAndSet(this, STATE_STARTED, STATE_SHUTDOWN)) {
            stateUpdater.set(this, STATE_SHUTDOWN);
        }
        return Collections.emptySet();
    }




}
